package com.gy.flink.project.data.producer;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class MyKafkaProducer {

    public static void main(String[] args) throws InterruptedException {

        Properties properties = new Properties();
        properties.put("bootstrap.servers", "centos103:9092");
        properties.put("key.serializer", StringSerializer.class);
        properties.put("value.serializer", StringSerializer.class);

        String topic = "allData";
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        while (true) {
            String message = getTestData();
            producer.send(new ProducerRecord<String, String>(topic, message));
            System.out.println(message);
            TimeUnit.SECONDS.sleep(3);
        }
//        producer.close();
    }

    private static String getTestData() {
        DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        HashMap<String, Object> map = new HashMap<>();
        map.put("dt", LocalDateTime.now().format(dtf));
        map.put("countryCode", getRandomAreas());
        ArrayList<Object> dataList = new ArrayList<>();
        map.put("data", dataList);
        for (int i = 0; i < getRandomInt(0, 10); i++) {
            HashMap<String, Object> dataMap = new HashMap<>();
            dataList.add(dataMap);

            dataMap.put("type", "s" + getRandomInt(1, 10));
            dataMap.put("score", Math.random());
            dataMap.put("level", String.valueOf(getRandomChar('A', 'G')));
        }

        return JSON.toJSONString(map);
    }

    public static int getRandomInt(int minNum, int maxNum) {
        int n = minNum + (int) (Math.random() * (maxNum - minNum));
        return n;
    }

    public static char getRandomChar(char startChar, char endChar) {
        char ch = (char) (startChar + Math.random() * (endChar - startChar + 1));
        return ch;
    }

    public static String getRandomAreas() {
        int randomInt = getRandomInt(0, 7);
        String areas = "IN";
        switch (randomInt) {
            case 0:
                areas = "US";
                break;
            case 1:
                areas = "TW";
                break;
            case 2:
                areas = "HK";
                break;
            case 3:
                areas = "PK";
                break;
            case 4:
                areas = "KW";
                break;
            case 5:
                areas = "SA";
                break;
            case 6:
                areas = "IN";
                break;
        }
        return areas;
    }
}
